package com.codejiwei.flink.sql;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/6/8
 * desc:
 **/
public class Flink_Table_SQL_API {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.createTemporaryTable("sourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10L)
                .build());

        tEnv.executeSql("create temporary table sinkTable with ('connector' = 'print') like sourceTable(excluding options)");

        Table table2 = tEnv.from("sourceTable");

        Table table3 = tEnv.sqlQuery("select * from sourceTable");

//        TableResult tableResult = table2.executeInsert("sinkTable");


        tEnv.executeSql("insert into sinkTable select * from sourceTable");
//        env.execute();

    }
}
